package cn.itcast.flink.base;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.concurrent.TimeUnit;

/**
 * Author itcast
 * Date 2021/8/1 16:50
 * Desc TODO
 */
public class StreamingFileSinkDemo {
    public static void main(String[] args) throws Exception {
        ParameterTool tools = ParameterTool.fromArgs(args);
        //1．初始化流计算运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2．设置Checkpoint（10s）周期性启动 和 stateBackend 存储路径
        env.enableCheckpointing(10000);
        env.setStateBackend(new FsStateBackend("file:///d:/chk"));
        //3．指定并行度为1
        env.setParallelism(1);
        //4．接入socket数据源，获取数据
        String hostname = tools.getRequired("hostname");
        int port = tools.getInt("port", 9999);
        DataStreamSource<String> source = env.socketTextStream(hostname, port);
        //5．创建Streamingfilesink对象
        //5-1. 创建输出文件配置，指定输出路径 /FlinkStreamFileSink/parquet
        String outputPath = "hdfs://node1:8020/FlinkStreamFileSink/parquet";
        //5-2.StreamingFileSink 行格式化 ， withBucketAssigner->DateTimeBucketAssigner
        //修改 输出文件的配置
        OutputFileConfig fileConfig = OutputFileConfig
                .builder()
                .withPartPrefix("netcat")
                .withPartSuffix(".txt")
            .build();

        StreamingFileSink<String> sink = StreamingFileSink
                .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
                //withRollingPolicy -> 默认滚筒策略
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withMaxPartSize(128 * 1024 * 1024)
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(1))
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(10))
                                .build()
                        //withOutputFileConfig -> 输出文件的配置
                ).withOutputFileConfig(fileConfig)
                .build();


        //6.设置输出 sink
        source.addSink(sink);
        //7．执行任务
        env.execute();
    }
}
